package com.thiswhyme.mq.kafka.consumer1.consumer;

import com.thiswhyme.mq.kafka.common.consts.TopicNameDefinition;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.stereotype.Component;

/**
 * @description: 消费者
 * @author: thiswhyme
 * @date: 2021/4/21
 */
@Slf4j
@Component
public class KafkaConsumer {

    /**
     * 监听普通消息1
     * @param record
     */
    @KafkaListener(topics = {TopicNameDefinition.COMMON_TOPIC})
    public void onCommonMessage1(ConsumerRecord<?, ?> record){
        log.info("=====【onCommonMessage1】监听到消息，topic：{}，partition：{}，内容：{}", record.topic(),
                record.partition(), record.value());
    }

    /**
     * 监听普通消息2
     * @param record
     */
    @KafkaListener(topics = {TopicNameDefinition.COMMON_TOPIC2})
    public void onCommonMessage2(ConsumerRecord<?, ?> record){
        log.info("=====【onCommonMessage2】监听到消息，topic：{}，partition：{}，内容：{}", record.topic(),
                record.partition(), record.value());
    }

    /**
     * 监听可过滤消息
     * @param record
     */
    @KafkaListener(topics = {TopicNameDefinition.FILTER_TOPIC}, containerFactory = "filterContainerFactory")
    public void onFilterMessage(ConsumerRecord<?, ?> record) {
        log.info("=====【onFilterMessage】监听到消息，内容：{}", record.value());
    }

    /**
     * 监听消息并转发
     * @param record
     * @return
     */
    @KafkaListener(topics = {TopicNameDefinition.RESEND_TOPIC})
    @SendTo(TopicNameDefinition.COMMON_TOPIC2)
    public String onResendMessage(ConsumerRecord<?, ?> record) {
        log.info("=====【onResendMessage】监听到消息，开始转发，内容：{}", record.value());
        return record.value() + "-被转发";
    }

    /**
     * 监听callback消息
     * @param record
     * @return
     */
    @KafkaListener(topics = {TopicNameDefinition.CALLBACK_TOPIC})
    public void onCallbackMessage(ConsumerRecord<?, ?> record) {
        log.info("=====【onCallbackMessage】监听到消息，内容：{}", record.value());
    }

    /**
     * 监听callback消息
     * @param record
     * @return
     */
    @KafkaListener(topics = {TopicNameDefinition.TRANSACTION_TOPIC})
    public void onTransactionMessage(ConsumerRecord<?, ?> record) {
        log.info("=====【onTransactionMessage】监听到消息，内容：{}", record.value());
    }

//    /**
//     * @Title 指定topic、partition、offset消费
//     * ① id：消费者ID；
//     * ② groupId：消费组ID；
//     * ③ topics：监听的topic，可监听多个；
//     * ④ topicPartitions：可配置更加详细的监听信息，可指定topic、parition、offset监听。
//     * 注意：topics和topicPartitions不能同时使用；
//     * @Description 同时监听topic1和topic2，监听topic1的0号分区、topic2的 "0号和1号" 分区，指向1号分区的offset初始值为8
//     * @Author long.yuan
//     * @Date 2020/3/22 13:38
//     * @Param [record]
//     * @return void
//     **/
//    @KafkaListener(id = "consumer1",groupId = "felix-group",topicPartitions = {
//            @TopicPartition(topic = "topic1", partitions = { "0" }),
//            @TopicPartition(topic = "topic2", partitions = "0", partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "8"))
//    })
//    public void onMessage2(ConsumerRecord<?, ?> record) {
//        System.out.println("topic:"+record.topic()+"|partition:"+record.partition()+"|offset:"+record.offset()+"|value:"+record.value());
//    }
//
//    /**
//     * # 设置批量消费
//     * spring.kafka.listener.type=batch
//     * # 批量消费每次最多消费多少条消息
//     * spring.kafka.consumer.max-poll-records=50
//     * @param records
//     */
//    @KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic1")
//    public void onMessage3(List<ConsumerRecord<?, ?>> records) {
//        System.out.println(">>>批量消费一次，records.size()="+records.size());
//        for (ConsumerRecord<?, ?> record : records) {
//            System.out.println(record.value());
//        }
//    }
//
//    // 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
//    @KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
//    public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
//        throw new Exception("简单消费-模拟异常");
//    }
//
//    // 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
//    @KafkaListener(topics = "topic1",errorHandler="consumerAwareErrorHandler")
//    public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
//        System.out.println("批量消费一次...");
//        throw new Exception("批量消费-模拟异常");
//    }
//
}